package com.xuef.dfs.datanode.server;

import com.xuef.dfs.namenode.rpc.model.HeartbeatRequest;
import com.xuef.dfs.namenode.rpc.model.HeartbeatResponse;
import com.xuef.dfs.namenode.rpc.model.RegisterRequest;
import com.xuef.dfs.namenode.rpc.model.RegisterResponse;
import com.xuef.dfs.namenode.rpc.service.NameNodeServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;

import java.util.concurrent.CountDownLatch;

public class NameNodeServiceActor {
    private String nameServerIp = "127.0.0.1";
    private int nameServerPort = 50070;
    private String datanodeIp = "127.0.0.1";
    private String datanodeHostname = "dfs-datanode-01";
//    NameNodeServiceGrpc.NameNodeServiceBlockingStub blockingStub;
//    ManagedChannel channel;

    public NameNodeServiceActor(){

    }

    public void start() {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        Thread registerThread = new RegisterThread(countDownLatch);
        registerThread.start();
        //注册完才会往下：启动心跳线程
        DataNodeHeartbeatThread dataNodeHeartbeatThread = new DataNodeHeartbeatThread();
        dataNodeHeartbeatThread.start();
    }

    /**
     * 注册线程
     */
    class RegisterThread extends Thread{
        CountDownLatch countDownLatch;
        public RegisterThread(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }

        public void register(){
            // Channel is the abstraction to connect to a service endpoint
            // Let's use plaintext communication because we don't have certs
            ManagedChannel channel = ManagedChannelBuilder.forTarget("localhost:50070")
                    .usePlaintext()
                    .build();
            // It is up to the client to determine whether to block the call
            // Here we create a blocking stub, but an async stub,
            // or an async stub with Future are always possible.
            NameNodeServiceGrpc.NameNodeServiceBlockingStub blockingStub = NameNodeServiceGrpc.newBlockingStub(channel);
            /** 从配置文件中获取 */
            RegisterRequest registerRequest = RegisterRequest.newBuilder()
                    .setIp("127.0.0.1").setHostname("dfs-datanode-01").build();

            // Finally, make the call using the stub
            RegisterResponse response = blockingStub.register(registerRequest);

            System.out.println(response);
            countDownLatch.countDown();
            // A Channel should be shutdown before stopping the process.
            channel.shutdownNow();
        }
        @Override
        public void run() {

            try {
                register();
                Thread.sleep(1000);
                countDownLatch.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * datanode 心跳维持线程
     */
    class DataNodeHeartbeatThread extends Thread{
        @Override
        public void run() {
            while(true){
                try {
                    Thread.sleep(10*1000);
                    heartbeat();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

        public void heartbeat(){
            ManagedChannel channel = ManagedChannelBuilder.forTarget("localhost:50070")
                    .usePlaintext()
                    .build();
            // It is up to the client to determine whether to block the call
            // Here we create a blocking stub, but an async stub,
            // or an async stub with Future are always possible.
            NameNodeServiceGrpc.NameNodeServiceBlockingStub blockingStub = NameNodeServiceGrpc.newBlockingStub(channel);
            /** 从配置文件中获取 */
            HeartbeatRequest registerRequest = HeartbeatRequest.newBuilder()
                    .setIp("127.0.0.1").setHostname("dfs-datanode-01").build();

            // Finally, make the call using the stub
            HeartbeatResponse response =
                    blockingStub.heartbeat(registerRequest);

            System.out.println("heartbeat response: " + response);

            // A Channel should be shutdown before stopping the process.
            channel.shutdownNow();
        }
    }
}
